package gossip.v1;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static gossip.v1.GossipConfig.*;

public  class GossipCore {
    private final MembershipTable membershipTable;
    private GossipTransport transport;
    private final FailureDetector failureDetector;
    private final String localNodeId;

    public GossipCore(int port) {
        this.membershipTable = new MembershipTable(port);
        this.failureDetector = new FailureDetector(this);
        this.localNodeId = "node-" + port;
    }

    public void setTransport(GossipTransport transport) {
        this.transport = transport;
    }

    public void start(ScheduledExecutorService scheduler) {
        scheduler.scheduleAtFixedRate(this::doGossip, 0, GOSSIP_INTERVAL, TimeUnit.MILLISECONDS);
        scheduler.scheduleAtFixedRate(failureDetector::check, 0, FAILURE_DETECTION_INTERVAL, TimeUnit.MILLISECONDS);
        System.out.printf("[CORE] Gossip scheduler started with %dms interval%n", GOSSIP_INTERVAL);
    }

    private void doGossip() {
        try {
            // 更新本地心跳
            membershipTable.incrementLocalHeartbeat();
            List<MemberState> targets = membershipTable.selectRandomTargets(MAX_GOSSIP_TARGETS);
            if (targets.isEmpty()) return;

            Map<String, Long> digest = new HashMap<>();
            membershipTable.getMembers().forEach((id, state) ->
                    digest.put(id, state.getHeartbeat()));

            for (MemberState target : targets) {
                sendGossipDigest(target.getAddress(), digest);
            }
        } catch (Exception e) {
            System.err.printf("[CORE] Gossip cycle failed: %s%n", e.getMessage());
        }
    }

    private void sendGossipDigest(InetSocketAddress target, Map<String, Long> digest) {
        String message = "DIGEST:" + serializeDigest(digest);
        transport.send(target, message);
    }

    private String serializeDigest(Map<String, Long> digest) {
        return digest.entrySet().stream()
                .map(e -> e.getKey() + "=" + e.getValue())
                .collect(Collectors.joining(","));
    }

    public void handleMessage(SocketAddress sender, String message) {
        try {
            if (message == null || message.isEmpty()) return;

            if (message.startsWith("DIGEST:")) {
                handleGossipDigest((InetSocketAddress) sender, message.substring(7));
            } else if (message.startsWith("SYNC:")) {
                handleSyncRequest((InetSocketAddress) sender, message.substring(5));
            } else if (message.startsWith("FULLSYNC:")) {
                handleFullSync(message.substring(9));
            } else if (message.startsWith("PROBE:")) {
                handleProbeRequest(message.substring(6));
            } else if (message.startsWith("ACK:")) {
                handleAck(message.substring(4));
            }
        } catch (Exception e) {
            System.err.printf("[CORE] Error processing message from %s: %s%n", sender, e.getMessage());
        }
    }

    // GossipCore.java
    private void handleGossipDigest(InetSocketAddress sender, String digestStr) {
        failureDetector.recordActivity(sender);
        Map<String, Long> remoteDigest = parseDigest(digestStr);
        Map<String, Long> localDigest = membershipTable.getMembersDigest();

        // 双向状态检测
        List<String> nodesToPush = new ArrayList<>();
        List<String> nodesToRequest = new ArrayList<>();

        // 1. 检测需要推送给发送方的节点
        for (String nodeId : localDigest.keySet()) {
            Long remoteHeartbeat = remoteDigest.get(nodeId);
            if (remoteHeartbeat == null || localDigest.get(nodeId) > remoteHeartbeat) {
                nodesToPush.add(nodeId);
            }
        }

        // 2. 检测需要从发送方请求的节点
        for (String nodeId : remoteDigest.keySet()) {
            Long localHeartbeat = localDigest.get(nodeId);
            if (localHeartbeat == null || remoteDigest.get(nodeId) > localHeartbeat) {
                nodesToRequest.add(nodeId);
            }
        }

        // 3. 主动推送状态
        if (!nodesToPush.isEmpty()) {
            Map<String, MemberState> statesToSend = new HashMap<>();
            for (String nodeId : nodesToPush) {
                MemberState state = membershipTable.getMembers().get(nodeId);
                if (state != null) statesToSend.put(nodeId, state);
            }
            String syncData = "FULLSYNC:" + serializeStates(statesToSend);
            transport.send(sender, syncData);
        }

        // 4. 请求需要同步的状态
        if (!nodesToRequest.isEmpty()) {
            String response = "SYNC:" + String.join(",", nodesToRequest);
            transport.send(sender, response);
        }
    }

    // GossipCore.java
    public void join(InetSocketAddress seedNode) {
        System.out.printf("[%s] Joining cluster via seed node %s%n", localNodeId, seedNode);

        // 创建包含本地状态的完整摘要
        Map<String, Long> digest = membershipTable.getMembersDigest();

        // 添加种子节点信息
        String seedNodeId = generateNodeId(seedNode);
        if (!digest.containsKey(seedNodeId)) {
            digest.put(seedNodeId, 1L); // 初始心跳值
        }

        sendGossipDigest(seedNode, digest);

        // 主动请求种子节点的完整状态
        transport.send(seedNode, "SYNC:ALL");
    }

    private void handleSyncRequest(InetSocketAddress sender, String nodeList) {
        System.out.printf("[%s] Received SYNC request from %s for nodes: %s%n",
                "", sender, nodeList);
        // 记录活跃状态
        failureDetector.recordActivity(sender);

        if (nodeList.isEmpty()) return;

        if ("ALL".equals(nodeList)) {
            // 发送完整状态
            String syncData = "FULLSYNC:" + serializeStates(membershipTable.getMembers());
            transport.send(sender, syncData);
        } else if (!nodeList.isEmpty()) {
            Map<String, MemberState> states = new HashMap<>();
            for (String nodeId : nodeList.split(",")) {
                MemberState state = membershipTable.getMembers().get(nodeId);
                if (state != null) {
                    states.put(nodeId, state);
                }
            }
            if (!states.isEmpty()) {
                String syncData = "FULLSYNC:" + serializeStates(states);
                transport.send(sender, syncData);
            }
        }

    }

    public static String generateNodeId(InetSocketAddress address) {
        return "node-" + address.getPort();
    }

    private void handleFullSync(String fullSyncData) {
        try {
            System.out.printf("[%s] Received FULLSYNC: %s%n",
                    "", fullSyncData);
            Map<String, MemberState> states = parseStates(fullSyncData);
            states.values().forEach(membershipTable::merge);
        } catch (Exception e) {
            System.err.printf("[CORE] Error processing FULLSYNC: %s%n", e.getMessage());
        }
    }

    private void handleProbeRequest(String payload) {
        String[] parts = payload.split(":");
        if (parts.length != 2) return;

        String suspectNodeId = parts[0];
        String requesterId = parts[1];

        MemberState state = membershipTable.getMembers().get(suspectNodeId);
        if (state != null) {
            transport.send(membershipTable.getMembers().get(requesterId).getAddress(),
                    "ACK:" + suspectNodeId);
        }
    }

    private void handleAck(String nodeId) {
        failureDetector.recordActivity(nodeId);
    }

    private Map<String, Long> parseDigest(String digestStr) {
        Map<String, Long> digest = new HashMap<>();
        if (digestStr == null || digestStr.isEmpty()) return digest;

        for (String part : digestStr.split(",")) {
            int idx = part.indexOf('=');
            if (idx > 0) {
                try {
                    String nodeId = part.substring(0, idx);
                    long heartbeat = Long.parseLong(part.substring(idx + 1));
                    digest.put(nodeId, heartbeat);
                } catch (NumberFormatException e) {
                    System.err.println("Invalid digest format: " + part);
                }
            }
        }
        return digest;
    }

    private String serializeStates(Map<String, MemberState> states) {
        StringBuilder sb = new StringBuilder();
        for (Map.Entry<String, MemberState> entry : states.entrySet()) {
            MemberState state = entry.getValue();
            sb.append(state.getNodeId()).append(":")
                    .append(state.getAddress().getHostName()).append(":")
                    .append(state.getAddress().getPort()).append(":")
                    .append(state.getHeartbeat()).append(":")
                    .append(serializeMetadata(state.getMetadata())).append(";");
        }
        return sb.toString();
    }

    private String serializeMetadata(Map<String, String> metadata) {
        return metadata.entrySet().stream()
                .map(e -> e.getKey() + "=" +
                        e.getValue().replace(";", "\\;")
                                .replace(",", "\\,")
                                .replace("=", "\\="))
                .collect(Collectors.joining(","));
    }

    // GossipCore.java
    private Map<String, MemberState> parseStates(String data) {
        Map<String, MemberState> states = new HashMap<>();
        if (data == null || data.isEmpty()) return states;

        String[] entries = data.split(";");
        for (String entry : entries) {
            if (entry.isEmpty()) continue;

            String[] parts = entry.split(":");
            if (parts.length < 5) continue;

            try {
                String nodeId = parts[0];
                // 使用IP地址而不是主机名，避免localhost/127.0.0.1问题
                String ip = parts[1].equals("localhost") ? "127.0.0.1" : parts[1];
                int port = Integer.parseInt(parts[2]);
                long heartbeat = Long.parseLong(parts[3]);
                Map<String, String> metadata = parseMetadata(parts[4]);

                InetSocketAddress address = new InetSocketAddress(ip, port);
                MemberState state = new MemberState(nodeId, address);
                state.setHeartbeat(heartbeat);
                state.setMetadata(metadata);
                states.put(nodeId, state);
            } catch (Exception e) {
                System.err.println("Error parsing state: " + entry + ", error: " + e.getMessage());
            }
        }
        return states;
    }

    private Map<String, String> parseMetadata(String metaStr) {
        Map<String, String> meta = new HashMap<>();
        if (metaStr == null || metaStr.isEmpty()) return meta;

        for (String pair : metaStr.split("(?<!\\\\),")) {
            String[] parts = pair.split("(?<!\\\\)=", 2);
            if (parts.length == 2) {
                String key = parts[0].replace("\\=", "=");
                String value = parts[1].replace("\\;", ";")
                        .replace("\\,", ",")
                        .replace("\\=", "=");
                meta.put(key, value);
            }
        }
        return meta;
    }

    public void updateLocalMetadata(String key, String value) {
        membershipTable.updateLocalMetadata(key, value);
    }

    public MembershipTable getMembershipTable() {
        return membershipTable;
    }

    public GossipTransport getTransport() {
        return transport;
    }

    public String getLocalNodeId() {
        return localNodeId;
    }
}
